package com.yjxxt.studentInfo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import java.io.IOException;

public class StudentInfoJob {
    private static String driverClass = "com.mysql.cj.jdbc.Driver";
    private static String url = "jdbc:mysql://192.168.88.101:3306/yjxxt?serverTimezone=GMT%2B8&characterEncoding=utf8&useUnicode=true&useSSL=false";
    private static String username = "root";
    private static String password = "Yjx#1234";
    private static String tableName = "student_info";
    private static String[] fields = {"id", "gender", "video_grade", "job_grade", "exam_grade", "total_grade"};

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        // 1. 配置文件
        Configuration configuration = new Configuration(true);
        configuration.set("mapreduce.framework.name","local");

        // 2. 开始载入数据库的配置文件
        DBConfiguration.configureDB(configuration, driverClass, url, username, password);

        // 3. 创建Job
        Job job = Job.getInstance(configuration);
        // 4. 设置Job
        job.setJobName("StudentInfoJob_"+System.currentTimeMillis());
        job.setJarByClass(StudentInfoJob.class);
        job.setNumReduceTasks(3);
        // 5. 设置读取和写出的HDFS地址
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(DBOutputFormat.class);
        // 6. 输入输出路径
        FileInputFormat.setInputPaths(job,new Path("/yjx/grade.txt"));
        DBOutputFormat.setOutput(job, tableName, fields);
        // 7. 数据格式
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        // 8. Mapper&Reducer
        job.setMapperClass(StudentInfoMapper.class);
        job.setReducerClass(StudentInfoReducer.class);
        // 9. 提交等待完成
        job.waitForCompletion(true);
    }
}
